package com.suning.sawp.service.impl.bi;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.annotation.Resource;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.suning.framework.dal.client.DalClient;
import com.suning.framework.image.common.FtpEntity;
import com.suning.framework.image.common.FtpUtils;
import com.suning.sawp.constants.LoggerConstants;
import com.suning.sawp.dto.bi.BiDealErrorException;
import com.suning.sawp.dto.bi.BiLineErrorNoticeException;
import com.suning.sawp.dto.bi.ColumnLimit;
import com.suning.sawp.dto.bi.ColumnLimitParams;
import com.suning.sawp.dto.bi.EntityFieldInfo;
import com.suning.sawp.intf.bi.BiDataTableRouteService;
import com.suning.sawp.intf.bi.BiFtpDataService;
import com.suning.sawp.intf.dao.BiExceptionLogDaoService;
import com.suning.sawp.intf.dao.UnDealBiFileLogDaoService;
import com.suning.sawp.po.bi.BiDataExceptionLog;
import com.suning.sawp.po.bi.UnDealBiFileLog;
import com.suning.sawp.service.util.ConfigUtil;
import com.suning.sawp.service.util.DateUtils;

public class BiFtpDataServiceImpl implements BiFtpDataService {
    
    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerConstants.SAWP_BI);
    
    private static final String SEPARATOR = "/";
    
    private static final String BI_BLANK = "-";
    
    private static final String BI_NULL_BLANK = "null";
    
    /**
     * 批量处理行上限数目
     */
    private static final int BATCH_LINE_NUM = 1000;
    
    /**
     * yyyy-MM-dd HH:mm:ss.SSS正则表达式
     */
    private static Pattern FULL_TIME_PATTERN = Pattern.compile("\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}.\\d{3}");
    
    /**
     * 科学计数法正则表达式
     */
    private static Pattern SCIENTIFIC_NOTATION = Pattern.compile("^((-?\\d+.?\\d*)[Ee]{1}(-?\\d+))$");
    
    /**
     * BI ftp地址
     */
    private String ftpAddress;
    
    /**
     * ftp 用户名
     */
    private String userName;
    
    /**
     * ftp 密码
     */
    private String password;
    
    /**
     * ftp 文件目录路径
     */
    private String ftpFileDirPath;
    
    /**
     * 本地临时文件目录
     */
    private String localTempFileDirPath;
    
    /**
     * BI DAT文件对应表相关联的entiy类名
     */
    private Map<String, String> entiyClazzs;
    
    /**
     * BI DAT文件对应表-sqlMap所在namespace
     */
    private Map<String, String> sqlNamespaces;
    
    @Resource
    DalClient dalClient;
    
    @Resource
    BiExceptionLogDaoService biExceptionLogDaoService;
    
    @Resource
    UnDealBiFileLogDaoService unDealBiFileLogDaoService;
    
    @Resource
    BiDataTableRouteService biDataTableRouteService;
    
    @Override
    public void dealBiFtpData(String biFileName, String statisDate) 
            throws BiDealErrorException {
    	LOGGER.debug("enter dealBiFtpData, " + biFileName + statisDate);
    	dealFtpData(statisDate, biFileName, false);
        LOGGER.debug("quit dealBiFtpData, " + biFileName + statisDate);
    }

    /***
     * 
     * 功能描述: 记录行数据处理异常日志<br>
     * 〈功能详细描述〉
     *
     * @param lineSum 总行记录数
     * @param errorLines 错误的行号记录列表
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选) 
     */
    private void logLineException(List<Integer> errorLines,
            String biFileName, String statisDate) throws BiLineErrorNoticeException {
        if (CollectionUtils.isNotEmpty(errorLines)) {
            // 考虑到异常日志表异常描述字段EXCEPTION_DESC varchar(300)字符长度超过290记录一条异常记录
            StringBuilder builder = new StringBuilder();
            for (int i = 0; i < errorLines.size(); i++) {
                builder.append(errorLines.get(i)).append(",");
                if (builder.length() > 290) {
                    // 满290个字符记录一条异常日志
                    BiDataExceptionLog exceptionLog = new BiDataExceptionLog();
                    exceptionLog.setBiFileName(biFileName);
                    exceptionLog.setStatisDate(statisDate);
                    exceptionLog.setExceptionType(BiDataExceptionLog.LINE_EXCEPTION);
                    exceptionLog.setExceptionDesc(builder.toString());
                    exceptionLog.setCreateTime(new Date());
                    biExceptionLogDaoService.addExceptionLog(exceptionLog);
                    builder = new StringBuilder();
                }
            }
            // 此份文件可能存在数据错乱
            // 此种情况说明，BI所抛数据有问题，添加到未处理异常文件表里，后面手动促发处理
            addUnDealFileLog(biFileName, statisDate, UnDealBiFileLog.UN_DEAL_NO_AUTO);
            throw new BiLineErrorNoticeException(BiLineErrorNoticeException.ERROR_MSG_LINE);
        }
    }
    
    /**
     * 
     * 功能描述: 新增未处理异常文件任务记录<br>
     * 〈功能详细描述〉
     *
     * @param biFileName
     * @param statisDate
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void addUnDealFileLog(String biFileName, String statisDate, String status) {
        UnDealBiFileLog unDealBiFileLog = new UnDealBiFileLog();
        unDealBiFileLog.setBiFileName(biFileName);
        unDealBiFileLog.setStatisDate(statisDate);
        unDealBiFileLog.setStatus(status);
        unDealBiFileLog.setCreateTime(new Date());
        unDealBiFileLog.setUpdateTime(new Date());
        unDealBiFileLogDaoService.addUnDealBiFileLog(unDealBiFileLog);
    }
    
    /**
     * 
     * 功能描述: 获取解析DAT数据入库所对应SQL字段<br>
     * 〈功能详细描述〉
     * entiy 成员变量顺序需要与BI DAT文件字段顺序保持，否则解析时会乱
     * @param clazz entiy所属成员变量需与SQL column字段保持一致
     * @return EntityFieldInfo 字段名：所属类型 以及 where条件字段集合
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private EntityFieldInfo getFieldNames(String className) {
        try {
            Class<?> clazz = Class.forName(className);
            Field[] fields = clazz.getDeclaredFields();
            Map<String, Class<?>> fieldMap = new LinkedHashMap<String, Class<?>>();
            Map<String, ColumnLimitParams> limitColumns = new HashMap<String, ColumnLimitParams>();
            // 表对应BEAN存在serialVersionUID和自增主键ID，跳过
            for (int i = 2; i < fields.length; i++) {
                fieldMap.put(fields[i].getName(), fields[i].getType());
                // 判断是否有字段限制条件
                if (fields[i].isAnnotationPresent(ColumnLimit.class)) {
                    ColumnLimit annotation = fields[i].getAnnotation(ColumnLimit.class);
                    // 字符长度限制
                    int varcharLen = annotation.length();
                    String columnName = fields[i].getName();
                    ColumnLimitParams limitParams = new ColumnLimitParams();
                    limitParams.setLength(varcharLen);
                    limitColumns.put(columnName, limitParams);
                }
            }
            
            return new EntityFieldInfo(fieldMap, limitColumns);
        } catch (Exception e) {
            LOGGER.error("反射获取entity:{}, 对应字段失败", className, e);
            return null;
        }
        
    }

    /**
     * 
     * 功能描述: 处理单行数据<br>
     * 〈功能详细描述〉
     *
     * @param vals BI DAT 文件的行记录数据
     * @param fieldInfo BI 表对应entity 反射所得表字段信息及where条件字段
     * @param biFileName 处理的BI文件名
     * @param currentLineNum 当前的行号
     * @param errorLineList 错误行号记录集合
     * @param goRewriteFlag 是否需要走重跑数据表
     * @return SQL执行的返回结果数组
     * @throws BiLineErrorNoticeException 
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void batchDealLineData(List<String[]> lineVals, EntityFieldInfo fieldInfo,
            String biFileName, int currentLineNum, List<Integer> errorLineList, boolean goRewriteFlag) throws BiLineErrorNoticeException {
    	LOGGER.debug("enter batchDealLineData, " + biFileName + "," + lineVals.size());
    	
        int[] rets = null;
        // 单行处理，若有异常直接吃掉处理，返回false，记录BI解析异常日志,不阻碍文件继续解析
        try {
            // 准备SQL所需参数
            List<Map<String, Object>> mapList = new ArrayList<Map<String,Object>>();
            Set<Entry<String, Class<?>>> entries = fieldInfo.getFields().entrySet();
            Map<String, ColumnLimitParams> limitColumns = fieldInfo.getLimitColumns();
            for (String[] vals : lineVals) {
                Map<String, Object> paramMap = new HashMap<String, Object>();
                int i = 0;
                for (Entry<String, Class<?>> entry : entries) {
                    Object obj = null;
                    if ("createTime".equals(entry.getKey())
                            || "updateTime".equals(entry.getKey())) {
                        obj = new Date();
                    } else {
                        obj = convertRelateVal(vals[i], entry.getValue()
                                , entry.getKey(), limitColumns);
                    }
                    paramMap.put(entry.getKey(), obj);
                    i++;
                }
                // 根据statisDate统计时间的月份选择当前应该入到哪张表中
                paramMap.put("goRewrite", goRewriteFlag);
                chooseTable(paramMap);
                mapList.add(paramMap);
            }
            
            LOGGER.debug(biFileName + ",mapList.size, " + mapList.size() + ", " + lineVals.size());
            
            //如果列表大于0,则进行批量处理
            if (CollectionUtils.isNotEmpty(mapList)){
            	
            	@SuppressWarnings("unchecked")
                Map<String, Object>[] batchValues = (Map<String, Object>[]) mapList.toArray(new Map[mapList.size()]);
                // sqlMap 对应SQL语句 id
                String sqlId = sqlNamespaces.get(biFileName) + ".UPSERT_FIELDS";
                
                // 执行SQL操作
                rets = dalClient.batchUpdate(sqlId, batchValues);
            } else {
                LOGGER.info("BI {}文件，当前行号{}，此批次数据为空", biFileName, currentLineNum);
            }
            
        } catch (Exception e) {
            LOGGER.error("批量解析BI文件数据出错", e);
            throw new BiLineErrorNoticeException(e.getMessage());
        }
        
        // 记录本批次中可能出现的失败行号
        if (null == rets) {
            // 此1000批次执行SQL失败
        	//说明是第一次批量处理
        	if (currentLineNum == lineVals.size()){
        		for (int i = 0; i < lineVals.size(); i++) {
                    errorLineList.add(i);
                }
        	}else{
        		//非首次,则所有行都出错
        		for (int i = 0; i < lineVals.size(); i++) {
                    errorLineList.add(currentLineNum + i);
                }
        	}
            
        } else {
        	//根据返回值大小判断插入,更新是否成功
            for (int i = 0; i < rets.length; i++) {
                if (rets[i] < 0) {
                    errorLineList.add(currentLineNum - rets.length + i);
                }
            }
        }
        
        // 清空列表准备承接下一批次数据
        lineVals.clear();
        LOGGER.debug("quit batchDealLineData, " + biFileName);
    }
    
    /**
     * 
     * 功能描述: 根据统计日期选择具体的分表<br>
     * 〈功能详细描述〉
     *
     * @param paramMap
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void chooseTable(Map<String, Object> paramMap) {
        // 选择统计数据的日期
        String statisDate = (String) paramMap.get("statisDate");
        Boolean goRewrite = (Boolean) paramMap.get("goRewrite");
        if (null == goRewrite) {
            goRewrite = false;
        }
        String index = biDataTableRouteService.optWriteGetTableIndex(statisDate, goRewrite);
        if (StringUtils.isNotBlank(index)) {
            // 获得有效的表编号
            paramMap.put("tableIndex", index);
        }
    }
    
    /**
     * 
     * 功能描述: 根据class类别将str转换为对应值<br>
     * 〈功能详细描述〉
     *
     * @param str
     * @param clazz
     * @param columnName
     * @param limitColumns
     * @return
     * @throws ParseException 
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private Object convertRelateVal(String str, Class<?> clazz, 
            String columnName, Map<String, ColumnLimitParams> limitColumns) throws ParseException {
        Object obj = null;
        ColumnLimitParams limitParams = null;
        if (null != limitColumns) {
            // 取该字段的限制条件
            limitParams = limitColumns.get(columnName);
        }
        // BI所给数据如果为空，以“-”（有部分包含“null”）标识，需要过滤之
        if (StringUtils.isNotBlank(str) 
                && !BI_BLANK.equals(str) && !BI_NULL_BLANK.equals(str)) {
            if (Double.class.equals(clazz)) {
                int pointIndex = str.indexOf(".");
                // BI给的数据有科学计数法，例如1.0E-4，得先判断此类数据,若是则转换为double浮点型
                Matcher matcher = SCIENTIFIC_NOTATION.matcher(str);
                if (matcher.matches()) {
                    BigDecimal bd = new BigDecimal(str);
                    str = bd.toPlainString();
                    // 此种数据需要为提成比例，需保留四位小数
                    if (pointIndex > 0
                            && ((str.length() - pointIndex) > 4)) {
                        // 存在小数点，且小数点后面小数超过4位，做截取操作
                        str = str.substring(0, pointIndex + 5);
                    }
                } else {
                    // BI给的数据有这种的28.799999999999997，和产品确认直接截取小数点后两位
                    if (pointIndex > 0
                            && ((str.length() - pointIndex) > 2)) {
                        // 存在小数点，且小数点后面小数超过两位，做截取操作
                        str = str.substring(0, pointIndex + 3);
                    }
                }
                
                obj = new Double(str);
            } else if (Integer.class.equals(clazz)) {
                obj = Integer.valueOf(str);
            } else if (Long.class.equals(clazz)) {
                obj = Long.valueOf(str);
            } else if (Date.class.equals(clazz)) {
                // 与BI确认过，所涉及到时间的为2015-08-30 16:01:35.686格式
                // 先做正则匹配，过着转换否则拒绝
                Matcher matcher = FULL_TIME_PATTERN.matcher(str);
                if (matcher.matches()) {
                    obj = DateUtils.getDateByStr(str, DateUtils.DEFAULY_FULL_TIME_PATTERN);
                }
            } else {
                obj = ensureDbLength(str, limitParams);
            }
        }
        
        return obj;
    }

    /**
     * 功能描述: 确保表对应字段长度<br>
     * 〈功能详细描述〉
     *
     * @param str
     * @param limitParams
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private Object ensureDbLength(String str, ColumnLimitParams limitParams) {
        Object obj = null;
        // String 类型
        if (null != limitParams && limitParams.getLength() > 0) {
            // 字段有限制，若超出需要做截取操作
            if (str.getBytes(Charsets.UTF_8).length > limitParams.getLength()) {
                // 字段值字节数大于数据库字段设定长度
                if (str.length() > limitParams.getLength()) {
                    str = str.substring(0, limitParams.getLength());
                }
                while (str.getBytes(Charsets.UTF_8).length <= limitParams.getLength()) {
                    str = str.substring(0, str.length() - 1);
                }
            }
        }
        obj = str;
        return obj;
    }


    @Override
    public void deleteBiFtpDataByDay(String biFileName, String statisDate) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("statisDate", statisDate);
        // 根据statisDate统计时间的月份选择当前应该入到哪张表中
        chooseTable(paramMap);
        // 根据FTP文件名和统计日期，选择SQL语句
        String sqlId = sqlNamespaces.get(biFileName) + ".DETELE_FIELDS";
        try {
            dalClient.execute(sqlId, paramMap);
        } catch (Exception e) {
            Object[] args = {biFileName, statisDate, e};
            LOGGER.error("删除{}，{}数据失败", args);
        }
    }
    
    public String getFtpAddress() {
        return ftpAddress;
    }

    public void setFtpAddress(String ftpAddress) {
        this.ftpAddress = ftpAddress;
    }

    public String getUserName() {
        return userName;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getFtpFileDirPath() {
        return ftpFileDirPath;
    }

    public void setFtpFileDirPath(String ftpFileDirPath) {
        this.ftpFileDirPath = ftpFileDirPath;
    }

    public String getLocalTempFileDirPath() {
        return localTempFileDirPath;
    }

    public void setLocalTempFileDirPath(String localTempFileDirPath) {
        this.localTempFileDirPath = localTempFileDirPath;
    }

    public Map<String, String> getEntiyClazzs() {
        return entiyClazzs;
    }

    public void setEntiyClazzs(Map<String, String> entiyClazzs) {
        this.entiyClazzs = entiyClazzs;
    }

    public Map<String, String> getSqlNamespaces() {
        return sqlNamespaces;
    }

    public void setSqlNamespaces(Map<String, String> sqlNamespaces) {
        this.sqlNamespaces = sqlNamespaces;
    }

    @Override
    public void dealBiIncDetailRecoverData(String statisDate, String biFileName) throws BiDealErrorException {
        // 处理提成明细重抛数据会涉及到重跑数据逻辑
        dealFtpData(statisDate, biFileName, true);
    }

    @Override
    public void dealBiIncomeRecoverData(String statisDate, String biFileName) throws BiDealErrorException {
        dealFtpData(statisDate, biFileName, false);
    }

    /**
     * 
     * 功能描述: 根据对应的BI文件名下载相应的BI文件<br>
     * 〈功能详细描述〉
     * 返回临时文件路径列表，.FLG文件放在第一个
     * @param statisDate
     * @param filename
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private List<String> downloadBiFtpFiles(String statisDate, String filename) {
        List<String> localTempPathList = null;
        // BI ftp地址、用户名、密码准备
        FtpEntity ftpEntity = new FtpEntity(ftpAddress, userName, password, null);
        FTPClient ftpClient = FtpUtils.borrowClient(ftpEntity);
        Map<String, String> biFileNames = new HashMap<String, String>();
        // ftp 上文件路径
        StringBuilder ftpBuilder = new StringBuilder(ftpFileDirPath).append(SEPARATOR)
                .append(statisDate).append(SEPARATOR);
        String ftpPrefix = ftpBuilder.toString();
        // 应用本地临时文件目录路径前缀
        StringBuilder tmpBuilder = new StringBuilder(localTempFileDirPath)
                .append(SEPARATOR);
        String tmpFilePathPrefix = tmpBuilder.toString();
        try {
            // 列出当前目录下的所有文件名称
            String[] ftpFilePaths = ftpClient.listNames(ftpPrefix);
            // 遍历目录下所有文件名，取符合filename前缀的所有文件
            for (String filePath : ftpFilePaths) {
                String[] args = filePath.split("/");
                String fileName = args[2];
                if (fileName.startsWith(filename)) {
                    // 应用本地临时文件路径
                    String tmpLocalFilePath = new StringBuilder(tmpFilePathPrefix).append(fileName).toString();
                    // ftp文件路径
                    biFileNames.put(filePath, tmpLocalFilePath);
                }
            }
            
            // 判断临时目录是否存在，不存在则创建
            File ftpTmpDir = new File(localTempFileDirPath);
            if (!ftpTmpDir.exists() &&
                    !ftpTmpDir.isDirectory()) {
                LOGGER.warn("downloadBiFtpData, {}临时目录不存在，创建之", localTempFileDirPath);
                ftpTmpDir.mkdir();
            }
            // 依次下载符合条件的文件
            Set<Entry<String, String>> entries = biFileNames.entrySet();
            // 下载到应用本地的临时文件路径集合，第一个放置.FLG文件的文件路径
            localTempPathList = new ArrayList<String>(entries.size());
            for (Entry<String, String> entry : entries) {
                LOGGER.info("downloadBiFtpData, FTP文件名称:{}", entry.getKey());
                // 下载信号完成文件
                FtpUtils.downFile(ftpEntity, entry.getKey(), entry.getValue());
                if (entry.getValue().endsWith(".FLG")) {
                    localTempPathList.add(0, entry.getValue());
                } else {
                    localTempPathList.add(entry.getValue());
                }
            }
        } catch (Exception e) {
            LOGGER.error("BiFtpDataService.downloadBiFtpData error", e);
            // 释放ftpClient资源
            FtpUtils.returnBrokenClient(ftpClient);
            ftpClient = null;
        } finally {
            // 释放ftpClient资源
            if (ftpClient != null) {
                FtpUtils.returnClient(ftpClient);
            }
        }
        
        return localTempPathList;
    }
    
    /**
     * 
     * 功能描述: 处理下载下来的BI文件<br>
     * 〈功能详细描述〉
     *
     * @param statisDate BI文件所处的日期目录
     * @param biFileName BI文件名称
     * @param goRewriteFlag 是否需要走月末重跑BI数据逻辑 true-走，false-不走
     * @throws BiDealErrorException
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void dealFtpData(String statisDate, String biFileName, boolean goRewriteFlag) throws BiDealErrorException {
        // 1、下载BI ftp文件数据文件集
        List<String> tempFilePaths = downloadBiFtpFiles(statisDate, biFileName);
        // 2、判断信号文件是否存在，如果存在则一次解析各个文件入库，否则往上抛出异常
        if (CollectionUtils.isNotEmpty(tempFilePaths)) {
            try {
                // 判断临时文件list的第一个文件是否为信号文件
                // 应用本地信号文件, 临时文件集的第一个为信号文件
                if (tempFilePaths.get(0).endsWith(".FLG") && 
                        new File(tempFilePaths.get(0)).exists()) {
                    // 取BI提成明细文件对应表相关联entity 字段信息
                    EntityFieldInfo fieldInfo = getFieldNames(entiyClazzs.get(biFileName));
                    // 依次解析处理各个文件入库
                    for (int i = 1; i < tempFilePaths.size(); i++) {
                        dealSingelFileData(tempFilePaths.get(i), biFileName, fieldInfo, goRewriteFlag);
                    }
                } else {
                    // 信号不存在
                    LOGGER.warn("dealFtpData, {}.FLG文件不存在", biFileName);
                    throw new BiDealErrorException(biFileName + ".FLG文件不存在");
                } 
            } catch (Exception e) {
                LOGGER.error("BiFtpDataService.dealFtpData deal error.", e);
                throw new BiDealErrorException(e.getMessage());
            } finally {
                // 不管BI文件处理成功与否，删除下载到应用的临时文件
                for (String tempFilePath : tempFilePaths) {
                    File tmpFlagFile = new File(tempFilePath);
                    if (tmpFlagFile.isFile() && tmpFlagFile.exists()) {
                        tmpFlagFile.delete();
                    }
                }
            }
            
        } else {
            LOGGER.warn("下载BI文件: {}失败", biFileName);
            throw new BiDealErrorException("下载BI文件" + biFileName + "失败");
        }
    }
    
    /**
     * 
     * 功能描述: 处理单个BI文件入库操作<br>
     * 〈功能详细描述〉
     *
     * @param localTempFilePath
     * @param biFileName
     * @param fieldInfo
     * @param goRewriteFlag
     * @throws BiDealErrorException
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void dealSingelFileData(String localTempFilePath, String biFileName,
            EntityFieldInfo fieldInfo, boolean goRewriteFlag) throws BiDealErrorException {
        BufferedReader br = null;
        File tempFile = null;
        FileInputStream fiStream = null;
        try {
            tempFile = new File(localTempFilePath);
            long fileSize = tempFile.length() / 1024 /1024;
            // 预设100M，超过上限直接抛异常，发告警短信
            if (fileSize > Integer.valueOf(ConfigUtil.getConfig("/config/sawp/config", "biFileMaxSize", "100"))) {
                throw new BiDealErrorException(localTempFilePath + ", 文件超出上限");
            }
            fiStream = new FileInputStream(tempFile);
            br = new BufferedReader(new InputStreamReader(fiStream, "UTF-8"));
            String line = null;
            // 当前处理的行号
            int currentLineNum = 0;
            // 文件总计行数
            int sum = 0;
            // 处理失败的行号记录列表
            List<Integer> errorLineList = new ArrayList<Integer>();
            // 待处理行数据列表
            List<String[]> lineVals = new ArrayList<String[]>();
            while (null != (line = br.readLine())) {
                // 当前行号、总行记录数+1
                currentLineNum++;
                
                // 去除行尾的行结束标识符|@|
                line = line.substring(0, line.length() - 3);
                // 按|$|分割各个字段值
                String[] vals = line.split("\\|\\$\\|");
                // 将行数据加入待处理行数据列表中
                lineVals.add(vals);
                // 提高性能，每隔1000条一个批次处理
                if (lineVals.size() == BATCH_LINE_NUM) {
                    // 批量处理
                    batchDealLineData(lineVals, fieldInfo, biFileName, currentLineNum, errorLineList, goRewriteFlag);
                }

                sum++;
            }
            // 文件行数不一定能刚好被1000整除，整个文件遍历完之后
            // 判断lineVals里面是否仍有数据，如果有再来一次批量处理,保证文件所有行数全部被处理掉
            if (!lineVals.isEmpty()){
                batchDealLineData(lineVals, fieldInfo, biFileName, currentLineNum, errorLineList, goRewriteFlag);
            }
            
            // 处理完成，如有行记录异常则记录到日常日志表中
            logLineException(errorLineList, biFileName, "");
            
            //如果文件内容为空,抛异常, 发短信提示
            if (sum == 0 || currentLineNum == 0){
                throw new BiDealErrorException(localTempFilePath + ", 文件内容为空");
            }
            
        } catch (Exception e) {
            LOGGER.error("解析DAT文件：{}失败", localTempFilePath, e);
            throw new BiDealErrorException("解析DAT文件: " + localTempFilePath, e);
        } finally {
            if (null != br) {
                try {
                    br.close();
                } catch (IOException e) {
                    LOGGER.error("释放br文件流失败", e);
                }
            }
            if (null != fiStream) {
                try {
                    fiStream.close();
                } catch (IOException e) {
                    LOGGER.error("释放fiStream文件流失败", e);
                }
            }
        }
    }
}
